/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.index.access;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.io.IOUtils;

import unisb.cs.core.binary.utils.BinaryUtils;
import unisb.cs.core.index.structure.CSSTreeWithLongOffsets;
import unisb.cs.core.index.structure.IndexHeaderWithLongOffsets;
import unisb.cs.data.tables.TableObjectProxy;
import unisb.cs.data.types.DataTypes;

/**
 * This class performs index lookup into a CSS-tree 
 * which stores record offsets as Long values.
 *
 */
public class IndexLookupWithLongOffsets extends IndexLookup{
	// CSS-tree with record offsets as Long values
	CSSTreeWithLongOffsets cssTree = null;

	long tupleOffset = -1;
	long contentSize = 0;
	public int indexAndHeaderSize;

	public IndexLookupWithLongOffsets(int key, int keyType, int keySize, String tableName) {
		super(key, keyType, keySize, tableName);
	}

	/**
	 * This method extracts the relevant data from the input split and 
	 * returns an iterator over the relevant records.
	 * 
	 * @param in The split input data stream
	 * @param lowKey The low key for index access
	 * @param highKey The high key for index access
	 * @param dataOffset The data offset to start index access from
	 * @return An iterator over the relevant records within this split. 
	 * @throws IOException
	 */
	public TableObjectProxyList extract(FSDataInputStream in, String lowKey, String highKey, long dataOffset) throws IOException {
		// Read Index header
		IndexHeaderWithLongOffsets idxHeader = new IndexHeaderWithLongOffsets(keyType, keySize);
		byte[] header = new byte[idxHeader.getHeaderSize()];
		IOUtils.readFully(in, header, 0, idxHeader.getHeaderSize());
		idxHeader.parse(header);
		indexAndHeaderSize = idxHeader.getHeaderSize() + idxHeader.getIndexSize();
		
		// Read index (only if required)
		readIndex(in, lowKey, highKey, idxHeader);
		
		if (tupleOffset >= 0) {
			// relevant data
			in.seek(dataOffset); // seek the starting of data
			IOUtils.skipFully(in, tupleOffset); // skip up to the desired leaf
			return new TableObjectProxyList(in, tupleOffset, contentSize, lowKey, highKey, tableName);
		} else
			return null;	// no relevant data found in the split
	}

	private void readIndex(FSDataInputStream in, String lowKey, String highKey, IndexHeaderWithLongOffsets idxHeader) throws IOException {
		// Read Index
		int overlapType = idxHeader.getOverlapType(lowKey, highKey);
		if (overlapType == IndexHeaderWithLongOffsets.SPANS || overlapType == IndexHeaderWithLongOffsets.RIGHT_CONTAINED) {
			// relevant data
			byte[] index = new byte[idxHeader.getIndexSize()];
			IOUtils.readFully(in, index, 0, idxHeader.getIndexSize());
			cssTree = new CSSTreeWithLongOffsets(index, idxHeader.getNofTuples(), 5, keySize, keyType, keySize, 0);
			tupleOffset = 0;
			contentSize = idxHeader.getLongDataSize();
		} else if (overlapType != IndexHeaderWithLongOffsets.NOT_CONTAINED) {
			// relevant data
			byte[] index = new byte[idxHeader.getIndexSize()];
			IOUtils.readFully(in, index, 0, idxHeader.getIndexSize());
			cssTree = new CSSTreeWithLongOffsets(index, idxHeader.getNofTuples(), 5, keySize, keyType, keySize, 0);
			tupleOffset = cssTree.get(lowKey);
			contentSize = idxHeader.getLongDataSize();
		} else {
			// no relevant data in the split
			tupleOffset = -1;
		}
	}
	

	/**
	 * This class iterates over the relevant records within the split.
	 * i.e. after locating the starting offset of the low key, we do a 
	 * sequential scan in the split till we reach the high key in the query. 
	 *
	 */
	public class TableObjectProxyList implements Iterator<TableObjectProxy> {
		long offset, splitSize;
		FSDataInputStream in;
		byte[] highKeyBytes;

		private byte[] data;
		
		/** variable to prefetch and store the next record */
		private TableObjectProxy nextTableObjectProxy = null;
		private boolean fetched = false;
		private String tableName;

		/**
		 * Create a record list.
		 * @param in The split data input stream
		 * @param offset The starting offset within the split
		 * @param splitSize The split size
		 * @param lowKey The low key in the query
		 * @param highKey The high key in the query
		 */
		public TableObjectProxyList(FSDataInputStream in, long offset, long splitSize, String lowKey, String highKey, String tableName) {
			this.offset = offset;
			this.splitSize = splitSize;
			this.in = in;
			this.tableName = tableName;

			highKeyBytes = getKeyBytes(highKey);
			nextTableObjectProxy = fetchNext();
			byte[] lowKeyBytes = getKeyBytes(lowKey);
			try {
				while (nextTableObjectProxy != null && cssTree.compareKeys(nextTableObjectProxy.getByteAttribute(key), lowKeyBytes) < 0)
					nextTableObjectProxy = fetchNext();
			} catch (Exception e) {
				e.printStackTrace();
			}
			fetched = true;
		}

		/**
		 * Convert key into byte array
		 */
		private byte[] getKeyBytes(String key) {
			switch (keyType) {
			case DataTypes.INT_TYPE:
				return BinaryUtils.toBytes(Integer.parseInt(key));
			case DataTypes.FLOAT_TYPE:
				return BinaryUtils.toBytes(Float.parseFloat(key));
			case DataTypes.CHAR_TYPE:
				return key.getBytes();
			case DataTypes.DATE_TYPE:
				return BinaryUtils.toBytes(key);
			default:
				return null;
			}
		}

		/**
		 * Do we have more relevant records within the split. 
		 */
		public boolean hasNext() {
			if (!fetched) {
				nextTableObjectProxy = fetchNext();
				fetched = true;
			}
			return nextTableObjectProxy == null ? false : true;
		}

		/**
		 * Fetch the next relevant record in the split.
		 */
		public TableObjectProxy next() {
			if (!fetched)
				nextTableObjectProxy = fetchNext();
			else
				fetched = false;
			return nextTableObjectProxy;
		}

		/**
		 * Skip the remaining data within the split.
		 * This method is useful when the split contains
		 * co-grouped data and we want to do index access 
		 * over one of the relations 
		 */
		public void skipRest() {
			try {
				if ((splitSize - offset) > 0)
					IOUtils.skipFully(in, splitSize - offset);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * skip one record
		 */
		public void remove() {
			fetchNext();
		}

		/**
		 * Read the next record from the input stream
		 * @return The next record
		 */
		private TableObjectProxy fetchNext() {
			if (offset < splitSize) {
				try {
					TableObjectProxy proxy = new TableObjectProxy(tableName);
					int recordLen = proxy.getSize();
					if (recordLen == 0)
						return null;
					if (offset > (splitSize - recordLen))
						return null;

					data = new byte[recordLen];
					IOUtils.readFully(in, data, 0, recordLen);
					offset += recordLen;
					proxy.setRecord(data);
					return (cssTree.compareKeys(proxy.getByteAttribute(key), highKeyBytes) > 0) ? null : proxy;
				} catch (Exception e) {
					e.printStackTrace();
				}

			}
			return null;
		}
	}
}
